package com.saibowisdom.storage.aspirator.es;

import com.alibaba.fastjson.JSON;
import com.saibowisdom.storage.aspirator.common.MetaRetrieve;
import com.saibowisdom.storage.aspirator.common.meta.FieldInfo;
import com.saibowisdom.storage.aspirator.common.meta.MonitorInfo;
import com.saibowisdom.storage.aspirator.common.meta.NodeInfo;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.AdminClient;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.health.ClusterIndexHealth;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.Serializable;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * Created by admin on 2017/1/11.
 */
public class ESAccessor implements Serializable {

    final static Logger logger = LoggerFactory.getLogger(ESAccessor.class);

    private static final long serialVersionUID = 1L;

    private Client client;
    private MetaRetrieve mr = null;
    /**
     * 表和分区映射
     */
    private Map<String, Integer> partitionMap = new HashMap<>();

    /**
     * 表和分区字段
     */
    private Map<String, String> partitionFieldMap = new HashMap<>();
    /**
     * index映射
     * 1存在
     */
    private Map<String, Integer> indexMap = new HashMap<>();

    private static Date LIMITDATE;

    static {
        try {
            LIMITDATE = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2000-01-01 00:00:00");
        } catch (ParseException e) {
            e.printStackTrace();
        }
    }


    /**
     * 初始化es client
     *
     * @param host    es地址
     * @param clusterName es集群名称
     */
    public void init(String host,String clusterName) {
        final Settings settings = Settings.settingsBuilder()
                .put("cluster.name", clusterName)
                .put("client.transport.sniff", true)
                .build();

        try {
            logger.info("真实地址是: "+host);
            logger.info("尝试获取地址是: "+InetAddress.getByName(host));
            client = TransportClient.builder()
                    .settings(settings).build()
                    .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), 9300));
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
    }


    public void setParameter(MetaRetrieve mr) {
        this.mr = mr;
    }



    public void createXContentBuilderToMapping(String indexName, String mappingName, Map<String, FieldInfo> fieldMap, int type) throws Exception {
        final XContentBuilder builder;
        try {
            new XContentFactory();
            builder = XContentFactory.jsonBuilder()
                    .startObject()
                    .startObject(mappingName)
                    .startObject("_source")
                    .field("compress", true)
                    //.field("enabled", false)
                    .endObject();
            if (type == 1) {
                builder.startObject("_ttl")
                        .field("enabled", true)
                        .field("default", "7d")
                        .endObject();
            }

            builder.startObject("properties");

            for (Map.Entry<String, FieldInfo> entry : fieldMap.entrySet()) {
                builder.startObject(entry.getValue().getField())
                        .field("type", entry.getValue().getDataTypeString("es"))
                        .field("store", entry.getValue().isStored() ? "yes" : "no");

                if (entry.getValue().getIndexed() == 1) {
                    builder.field("index", "not_analyzed");
                }

                if (entry.getValue().getIndexed() == 2) {
                    builder.field("indexAnalyzer", "ik");
                    builder.field("searchAnalyzer", "ik");
                }

                builder.endObject();
            }

            builder.startObject("indextime")
                    .field("type", "date")
                    .field("store", "no");
            builder.endObject();

            builder.endObject()
                    .endObject()
                    .endObject();


        } catch (IOException e) {
            // TODO Auto-generated catch block
            throw new Exception("构造mapping " + indexName + " 失败,失败信息 " + e.getMessage());
        }

        final PutMappingRequest mapping = Requests.putMappingRequest(indexName).type(mappingName).source(builder);
        try {
            client.admin().indices().putMapping(mapping).actionGet();
        } catch (ElasticsearchException e) {
            throw new Exception("更新mapping " + indexName + " 失败,失败信息 " + e.getMessage());
        }
    }



    /**
     * 创建mapping
     *
     * @param tableName 表名
     * @param dataMap   数据map
     * @param isChange  是否有改变
     * @throws Exception 异常
     */

    public synchronized void createMapping(String tableName, Map<String, Object> dataMap, boolean isChange) throws Exception {

        final String[] db = tableName.split("$");
        final Map<String, FieldInfo> fieldMap = mr.getFieldTypes(tableName);
        String indexName = "";
        String table = "";
        if (db.length < 2) {
            indexName = table = tableName;
        }
        if (db.length >= 2) {
            indexName = db[0];
            table = db[1];
        }


        //更新操作
        if (isChange) {
            Map<String, ClusterIndexHealth> map = null;
            try {
                map = client.admin().cluster()
                        .health(new ClusterHealthRequest(indexName + "*")).actionGet()
                        .getIndices();
            } catch (ElasticsearchException e1) {
                e1.printStackTrace();
            }

            for (String index : map != null ? map.keySet() : new HashMap<String, ClusterIndexHealth>().keySet()) {
                if (index.equals(indexName + "-cache")) {
                    //缓存库
                    createXContentBuilderToMapping(index, tableName, fieldMap, 1);
                } else {
                    createXContentBuilderToMapping(index, tableName, fieldMap, 0);
                }
            }
        } else {
            /** indexName是否存在*/
        /*final Map map;
        try {
            map = client.admin().cluster()
                    .health(new ClusterHealthRequest(indexName)).actionGet()
                    .getIndices();
        } catch (ElasticsearchException e1) {
            throw new Exception("获取 " + indexName + " 是否存在失败,失败信息 " + e1.getMessage());
        }*/


            /**
             * 分区字段
             */
            String partitionField = partitionFieldMap.get(tableName) == null ? "" : partitionFieldMap.get(tableName);
            //建立分库 0-不分区,1-按天分区,2-按月分区,3-按年分区
            int partitionType = 0;
            if (partitionMap.get(tableName) != null) {
                partitionType = partitionMap.get(tableName);
            } else {
                for (String key : fieldMap.keySet()) {
                    partitionType = fieldMap.get(key).getPartitionType();
                    if (partitionType > 0) {
                        partitionField = key;
                        partitionFieldMap.put(tableName, key);
                        break;
                    }
                }
                partitionMap.put(tableName, partitionType);
            }


            final String cacheIndexName = indexName + "-cache";
            //缓存库
            if (indexMap.get(cacheIndexName) == null) {
                AdminClient ac = client.admin();
                final boolean cacheexists = ac.indices().exists(new IndicesExistsRequest(cacheIndexName)).actionGet().isExists();
                if (!cacheexists) {
                    client.admin().indices()
                            .create(new CreateIndexRequest(cacheIndexName)).actionGet();
                    client.admin()
                            .cluster()
                            .health(new ClusterHealthRequest(cacheIndexName)
                                    .waitForYellowStatus()).actionGet();
                    createXContentBuilderToMapping(cacheIndexName, tableName, fieldMap, 1);
                    indexMap.put(cacheIndexName, 1);
                }
            }

            /**
             * 是否可以建库
             */
            boolean isCreat = true;
            //分区设置
            if (partitionType > 0) {

                /** 是否有日期字段 */
                boolean hasDate = false;
                for (Map.Entry<String, Object> entry : dataMap.entrySet()) {
                    final MetaRetrieve.DataType dtt = mr.getType(table, "hbase", entry.getKey());

                    /** 日期字段处理 */
                    if (dtt == MetaRetrieve.DataType.DATE) {

                        /** 该字段是分区字段 */
                        if (entry.getKey().equals(partitionField)) {
                            final String newvalue = mr.dateTrans(entry.getValue().toString().trim());
                            final Date now = new Date();
                            final Date value = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(newvalue);
                            //如果日期在2000年以前或者在当前时间之后则不建库
                            if (value.before(LIMITDATE) || value.after(now)) {
                                isCreat = false;
                                break;
                            }

                            hasDate = true;
                            if (newvalue == null) {
                                //indexName = indexName + "-nodate";
                                if (partitionType == 1) {
                                    indexName += "-" + String.valueOf(new SimpleDateFormat("yyyyMMdd").format(new Date()));
                                } else if (partitionType == 2) {
                                    indexName += "-" + String.valueOf(new SimpleDateFormat("yyyyMM").format(new Date()));
                                } else if (partitionType == 3) {
                                    indexName += "-" + String.valueOf(new SimpleDateFormat("yyyy").format(new Date()));
                                }
                            } else {
                                if (partitionType == 1) {
                                    indexName += "-" + String.valueOf(new SimpleDateFormat("yyyyMMdd").format(value));
                                } else if (partitionType == 2) {
                                    indexName += "-" + String.valueOf(new SimpleDateFormat("yyyyMM").format(value));
                                } else if (partitionType == 3) {
                                    indexName += "-" + String.valueOf(new SimpleDateFormat("yyyy").format(value));
                                }
                            }
                            break;
                        }

                    }

                }

                /** 没有分区日期字段 补当前时间*/
                if (!hasDate) {
                    if (partitionType == 1) {
                        indexName += "-" + String.valueOf(new SimpleDateFormat("yyyyMMdd").format(new Date()));
                    } else if (partitionType == 2) {
                        indexName += "-" + String.valueOf(new SimpleDateFormat("yyyyMM").format(new Date()));
                    } else if (partitionType == 3) {
                        indexName += "-" + String.valueOf(new SimpleDateFormat("yyyy").format(new Date()));
                    }
                }

            } else {
                /** 是否有日期字段 */
                for (Map.Entry<String, Object> entry : dataMap.entrySet()) {
                    final MetaRetrieve.DataType dtt = mr.getType(table, "es", entry.getKey());

                    /** 日期字段处理 */
                    if (dtt == MetaRetrieve.DataType.DATE) {
                        if (entry.getKey().toLowerCase().equals("createdate")) {
                            final String newvalue = mr.dateTrans(entry.getValue().toString().trim());
                            final Date now = new Date();
                            final Date value = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(newvalue);
                            //如果日期在2000年以前或者在当前时间之后则不建库
                            if (value.before(LIMITDATE) || value.after(now)) {
                                isCreat = false;
                                break;
                            }
                        }

                    }
                }
            }

            if (indexMap.get(indexName) == null && isCreat) {
                final boolean fkexists = client.admin().indices().exists(new IndicesExistsRequest(indexName)).actionGet().isExists();

                if (!fkexists) {
                    client.admin().indices()
                            .create(new CreateIndexRequest(indexName)).actionGet();
                    client.admin()
                            .cluster()
                            .health(new ClusterHealthRequest(indexName)
                                    .waitForYellowStatus()).actionGet();
                    createXContentBuilderToMapping(indexName, tableName, fieldMap, 0);
                    indexMap.put(indexName, 1);
                }
            }


        }


        //return true;

    }

    /**
     * 移除内容中的html标签
     *
     * @param inputString 输入内容
     * @return string
     * @throws Exception 异常
     */
    public String htmlRemoveTag(String inputString) throws Exception {
        if (inputString == null) {
            return null;
        }
        String htmlStr = inputString; // 含html标签的字符串
        final String textStr;
        final Pattern p_script;
        final Matcher m_script;
        final Pattern p_style;
        final Matcher m_style;
        final Pattern p_html;
        final Matcher m_html;
        try {
            //定义script的正则表达式{或<script[^>]*?>[\\s\\S]*?<\\/script>
            final String regEx_script = "<[\\s]*?script[^>]*?>[\\s\\S]*?<[\\s]*?\\/[\\s]*?script[\\s]*?>";
            //定义style的正则表达式{或<style[^>]*?>[\\s\\S]*?<\\/style>
            final String regEx_style = "<[\\s]*?style[^>]*?>[\\s\\S]*?<[\\s]*?\\/[\\s]*?style[\\s]*?>";
            final String regEx_html = "<[^>]+>"; // 定义HTML标签的正则表达式
            p_script = Pattern.compile(regEx_script, Pattern.CASE_INSENSITIVE);
            m_script = p_script.matcher(htmlStr);
            htmlStr = m_script.replaceAll(""); // 过滤script标签
            p_style = Pattern.compile(regEx_style, Pattern.CASE_INSENSITIVE);
            m_style = p_style.matcher(htmlStr);
            htmlStr = m_style.replaceAll(""); // 过滤style标签
            p_html = Pattern.compile(regEx_html, Pattern.CASE_INSENSITIVE);
            m_html = p_html.matcher(htmlStr);
            htmlStr = m_html.replaceAll(""); // 过滤html标签
//	        textStr = htmlStr.replace(" ","");
            textStr = htmlStr;
        } catch (Exception e) {
            throw new Exception("移除html标签失败,错误信息  " + e.getMessage());
        }
        return textStr;// 返回文本字符串
    }

    /**
     * 插入数据到elasticsearch
     *
     * @param table 索引
     * @param objs  数据
     * @param lstPk 主键
     * @return int 0 正确 1 数据错误 需跳过  2 系统错误 等待执行
     * @throws Exception 异常
     */
    public synchronized int insert(String table, Map<String, Object> objs, List<String> lstPk) throws Exception {
        final String json;
        final String[] db = table.split("$");

        String index = null;
        String mapping = null;
        if (db.length < 2) {
            index = mapping = table;
        }

        if (db.length >= 2) {
            index = db[0];
            mapping = db[1];
        }


        String dateIndex = index;
        final Iterator<Map.Entry<String, Object>> itor = objs.entrySet().iterator();
        final StringBuilder key = new StringBuilder("");
        for (final String str : lstPk) {
            final String pk = objs.get(str).toString();
            if (pk == null) {
                logger.warn("数据 {} 没有主键{}", objs, str);
                return 1;
            }
            key.append(objs.get(str)).append(":");
        }
        if(key.length()>2)
            key.deleteCharAt(key.length() - 1);
        logger.info("当前处理数据主键为 : " + key.toString());
        final XContentBuilder contentBuilder;
        try {
            contentBuilder = XContentFactory.jsonBuilder()
                    .startObject();
        } catch (IOException e) {
            throw new Exception("contentBuilder创建失败,错误信息 " + e.getMessage());
        }
        //final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        //分区字段
        final String partitionField = partitionFieldMap.get(table) == null ? "" : partitionFieldMap.get(table);
        /** 有日期 */
        boolean hasDate = false;
        /** 是否要抛弃数据 */
        boolean cast = false;
        final int partitionType = partitionMap.get(table);
        while (itor.hasNext()) {
            final Map.Entry<String, Object> entry = itor.next();

            final MetaRetrieve.DataType dtt = mr.getType(table, "es", entry.getKey());

            if (dtt == null) {
                logger.warn("字段 {} 没有格式,放弃此字段 ", entry.getKey());
                continue;
            }

            String newvalue = entry.getValue().toString();

            /** 截取长度 */
            if (newvalue.length() > 1000 && newvalue.contains("<")) {
                newvalue = htmlRemoveTag(newvalue);
            }
            if (newvalue.length() > 10000) {
                newvalue = newvalue.substring(0, 10000);
            }

            /** 日期字段处理 */
            if (dtt == MetaRetrieve.DataType.DATE) {
                newvalue = mr.dateTrans(newvalue.trim());
                if (newvalue == null) {
                    //日期错误则赋予当前时间
                    newvalue = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
                }

                try {
                    final Date value = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(newvalue);

                    /** 日期字段是分区字段 */
                    if (entry.getKey().equals(partitionField)) {

                        //如果日期在2000年以前或者在当前时间之后则存入数据库
                        if (value.before(LIMITDATE) || value.after(new Date())) {
                            try {
                                //mr.insertErrorData(key.toString(), newvalue);
                                insertErrorData(key.toString(), newvalue);
                            } catch (Exception e) {
                                logger.error("日期区间问题插入mysql失败 {} ", e.getMessage());
                            }
                            cast = true;
                            break;
                        }

                        hasDate = true;
                        if (partitionType > 0) {
                            if (partitionType == 1) {
                                dateIndex = index + "-" + String.valueOf(new SimpleDateFormat("yyyyMMdd").format(value));
                            } else if (partitionType == 2) {
                                dateIndex = index + "-" + String.valueOf(new SimpleDateFormat("yyyyMM").format(value));
                            } else if (partitionType == 3) {
                                dateIndex = index + "-" + String.valueOf(new SimpleDateFormat("yyyy").format(value));
                            }
                        }
                    } else if (entry.getKey().toLowerCase().equals("createdate")) {
                        //如果日期在2000年以前或者在当前时间之后则存入数据库
                        if (value.before(LIMITDATE) || value.after(new Date())) {
                            try {
                                //mr.insertErrorData(key.toString(), newvalue);
                                insertErrorData(key.toString(), newvalue);
                            } catch (Exception e) {
                                logger.error("日期区间问题插入mysql失败 {} ", e.getMessage());
                            }
                            cast = true;
                            break;
                        }
                    }

                    contentBuilder.field(entry.getKey(), String.valueOf(value.getTime()));
                } catch (ParseException e) {
                    contentBuilder.field(entry.getKey(), String.valueOf(new Date().getTime()));

                } catch (IOException e) {
                    throw new Exception(entry.getKey() + " contentBuilder.field失败,错误信息  " + e.getMessage());
                }
            } else {
                try {
                    contentBuilder.field(entry.getKey(), newvalue + "");
                } catch (IOException e) {
                    throw new Exception(entry.getKey() + " contentBuilder.field失败,错误信息  " + e.getMessage());
                }
            }

        }

        try {
            contentBuilder.field("indextime", String.valueOf(new Date().getTime()));
        } catch (IOException e) {
            throw new Exception("生成indextime失败,错误信息  " + e.getMessage());
        }

        /** 没有分区日期字段 补当前时间*/
        if (!hasDate) {
            if (partitionType > 0) {
                contentBuilder.field(partitionField, String.valueOf(new Date().getTime()));
                if (partitionType == 1) {
                    dateIndex = index + "-" + String.valueOf(new SimpleDateFormat("yyyyMMdd").format(new Date()));
                } else if (partitionType == 2) {
                    dateIndex = index + "-" + String.valueOf(new SimpleDateFormat("yyyyMM").format(new Date()));
                } else if (partitionType == 3) {
                    dateIndex = index + "-" + String.valueOf(new SimpleDateFormat("yyyy").format(new Date()));
                }
            }
        }

        try {
            json = contentBuilder.endObject().string();
        } catch (IOException e) {
            throw new Exception("contentBuilder构建失败,错误信息  " + e.getMessage());
        }

        if (!cast) {
            boolean insert = false;
            try {
                client.prepareUpdate(index + "-cache", mapping, key.toString()).setDoc(json).execute()
                        .actionGet();
            } catch (Exception e) {
                if (e.getMessage().contains("document missing")) {
                    insert = true;
                } else {
                    throw new Exception("更新数据到" + index + "-cache失败,id为" + key + ",错误信息  " + e.getMessage());
                }
            }

            if (!insert) {
                try {
                    client.prepareUpdate(dateIndex, mapping, key.toString()).setDoc(json).execute()
                            .actionGet();
                } catch (Exception e) {
                    if (!e.getMessage().contains("document missing")) {
                        throw new Exception("更新数据到" + dateIndex + "失败,id为" + key + ",错误信息  " + e.getMessage());
                    }
                }
            }


            if (insert) {
                //插入7天缓存库
                try {
                    client.prepareIndex(index + "-cache", mapping, key.toString()).setSource(json).execute()
                            .actionGet();
                } catch (ElasticsearchException e) {
                    throw new Exception("插入数据到ES缓存库失败,错误信息  " + e.getMessage());
                }


                //插入数据库
                try {
                    client.prepareIndex(dateIndex, mapping, key.toString()).setSource(json).execute()
                            .actionGet();
                } catch (ElasticsearchException e) {
                    throw new Exception("插入数据到ES日期库失败,错误信息  " + e.getMessage());
                }
            }

        }

        return 0;
    }


    /**
     * 插入错误数据
     * 考虑是否用返回值的方式传递插入数据是否成功?
     * @return
     */
    public void insertErrorInfo(String ip, String host, String content, String error, int etype, int source) throws Exception {
        final MonitorInfo monitorInfo = new MonitorInfo();
        monitorInfo.setContent(content);
        monitorInfo.setError(error);
        monitorInfo.setHost(host);
        monitorInfo.setIp(ip);
        monitorInfo.setType(etype);
        monitorInfo.setSource(source);
        final String jsonString = JSON.toJSONString(monitorInfo);
        IndexResponse response;
        try {
            response = client.prepareIndex("aspirator_errordata","info").setSource(jsonString).execute().actionGet();
        } catch (ElasticsearchException e){
            throw new Exception("插入错误数据:"+e.getMessage());
        }
        //boolean flag = response.isCreated();
    }

    public void insertErrorData(String key, String errorData) throws Exception {
        Map<String, String> errorMap = new HashMap<>();
        errorMap.put("errorkey", key);
        errorMap.put("errorData", errorData);
        try {
            client.prepareIndex("aspirator_errorMapdata","info").setSource(errorMap).execute().actionGet();
        } catch (ElasticsearchException e){
            throw new Exception("插入错误数据:"+e.getMessage());
        }
    }

    /**
     * 查询插入主库时的错误数据
     * @param nodeInfo
     * @return
     */
    public ArrayList<MonitorInfo> getErrorData(NodeInfo nodeInfo){
        SearchResponse response = client.prepareSearch("aspirator_errordata")
                .setTypes("info")
                .setSearchType(SearchType.QUERY_THEN_FETCH)
                .setQuery(QueryBuilders.boolQuery().must(QueryBuilders.termQuery("dealflag",1))
                        .must(QueryBuilders.termQuery("ip",nodeInfo.getIp()))
                        .must(QueryBuilders.termQuery("host",nodeInfo.getHost())))
                .setSize(1000)
                .execute()
                .actionGet();

        SearchHit[] hitArr = response.getHits().getHits();

        final ArrayList<MonitorInfo> monitorInfoList = new ArrayList<>();

        if (hitArr!=null&&hitArr.length>0){
            for (int i = 0; i<hitArr.length; i++){
                Map<String, Object> source = hitArr[i].getSource();

                MonitorInfo monitorInfo = new MonitorInfo();
                String data = (String) source.get("errordata");
                monitorInfo.setId(Integer.parseInt(hitArr[i].getId()));
                monitorInfo.setCount(hitArr.length);
                monitorInfo.setIp(source.get("ip").toString());
                monitorInfo.setHost(source.get("host").toString());
                monitorInfo.setContent(source.get("content").toString());

            }
        }
        return monitorInfoList;
    }
}
